package Task;

import java.io.Serializable;
import java.util.ArrayList;
import java.util.ConcurrentModificationException;
import java.util.Enumeration;
import java.util.Hashtable;
import java.util.Iterator;
import java.util.Set;

import worker.Worker;

/**
 * Impl�mentation de la m�thode r�duce.
 * 
 * C'est ici qu'est stock� la hashtable.
 * @author fifi
 *
 */
public class ReduceCompteMots implements Serializable, Reduce {

	private static final long serialVersionUID = 1L;
	private Hashtable<String, Integer> table ;

	public ReduceCompteMots() {
		this.table = new Hashtable<String, Integer>();
	}
	
	/**
	 * Check l'existance d'un mot dans la hashtable locale (table).
	 * @return : true si le mot existe dans la hashtable et false sinon.
	 */
	@Override
	public synchronized boolean checkMot(String mot) {

		Set<String> set = table.keySet();
	    Iterator<String> itr = set.iterator();
		
	    while (itr.hasNext()) {
	    	
	    	String motTable = itr.next();
	    	
	    	if( motTable.compareTo(mot) == 0)
			{
	    		return true;
			} 
	    }
		return false;
	}

	@Override
	public synchronized Hashtable<String, Integer> getHashtable() {
		
		return this.table;
	}

	@Override
	public synchronized void setHashtable(Hashtable<String, Integer> table) {
			
		this.table = table;
	}

	/**
	 * 
	 * @return : la liste des mots de la hashtable (table).
	 */
	@Override
	public ArrayList<String> getListMots() {
		
		ArrayList<String> listMot = new ArrayList<String>();
		
		Set<String> set = table.keySet();
	    Iterator<String> itr = set.iterator();
		
	    while (itr.hasNext()) {
			
	    	listMot.add( itr.next() );
	    }
	    
	    return listMot;
	}

	/**
	 * Ajoute une occurence d�ja existante dans la hashtable.
	 */
	@Override
	public void ajouterOccurence(String mot, int nb) {
		
		synchronized(table) {
			if( table.containsKey( mot ) )
			{
				table.put(mot, table.get(mot) + nb );
			} else {
				table.put(mot, nb );
			}
		}
	}

	/**
	 * Methode reduce.
	 * Elle check l'existance des mots dans les hahstable des autres workers.
	 * 
	 * 	Si le mot existe ailleurs, l'occurence est ajout� sur l'autre workers et elle est supprim�e en locale.
	 * 
	 *  exemple:
	 *  	soit le mot i;
	 *  	si il existe chez le worker j;
	 *  	{
	 *  		j.getReduce().ajouterOccurence( i , nb(i) );
	 *  		remove(i);
	 *  	}	
	 *  	
	 *  
	 *  Au final, chaque worker aura la responsabilit� de mots ( nbMots E [0 ; nbMotsMax] ).
	 */
	@Override
	public void reduce( ArrayList<Worker> listWorkers ) {

		synchronized(table) {
			try{
			    	Enumeration<String> iter = table.keys();
			    	while (iter.hasMoreElements()) {
				    	
			    		String motTable = (String) iter.nextElement();
				    	boolean trouve = false;
				    	int i = 0;
						while( i < listWorkers.size() && !trouve) {
							
							if( listWorkers.get(i).getReduce().checkMot(motTable) ) {
								listWorkers.get(i).getReduce().ajouterOccurence(motTable, table.get(motTable));
								trouve = true;
								
								table.remove(motTable);
							}
							
							i++;
						}
					} 
				}
			catch (ConcurrentModificationException e) {
				System.out.println("Erreur : " + e);
			} 
			
		}
	}  
	
	/**
	 * Methode reduce.
	 * Elle check l'existance des mots dans les hahstable des autres workers.
	 * 
	 * 	Si le mot existe ailleurs, l'occurence est ajout� sur l'autre workers et elle est supprim�e en locale.
	 * 
	 *  exemple:
	 *  	soit le mot i;
	 *  	si il existe chez le worker j;
	 *  	{
	 *  		j.getReduce().ajouterOccurence( i , nb(i) );
	 *  		remove(i);
	 *  	}	
	 *  	
	 *  
	 *  Au final, chaque worker aura la responsabilit� de mots ( nbMots E [0 ; nbMotsMax] ).
	 */
	@Override
	public void reduce( ArrayList<Worker> listWorkers , int monRang ) {

		synchronized(table) {
			try{
			    	Enumeration<String> iter = table.keys();
			    	while (iter.hasMoreElements()) {
				    	
			    		String motTable = (String) iter.nextElement();
				    	boolean trouve = false;
				    	int i = monRang;
						while( i < listWorkers.size() && !trouve) {
							
							if( monRang != i ) {
								if( listWorkers.get(i).getReduce().checkMot(motTable) ) {
									
									listWorkers.get(i).getReduce().ajouterOccurence(motTable, table.get(motTable));
									trouve = true;
									
									table.remove(motTable);
								}
							}
							
							i++;
						}
					} 
				}
			catch (ConcurrentModificationException e) {
				System.out.println("Erreur : " + e);
			} 
			
		}
	}  
}
